package com.xzr.demo.mqserver.datacenter;

import com.xzr.demo.common.BinaryTool;
import com.xzr.demo.common.MqException;
import com.xzr.demo.mqserver.core.MSGQueue;
import com.xzr.demo.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 *通过这个类， 针对硬盘上的消息进行管理
 *
 * @Author：徐志荣
 * @Date：2023/8/8 下午 7:03
 */
public class MessageFileManager {
    //定义一个静态内部类，来表示该队列的统计信息
    static public  class Stat{
        public int totalCount; // 总消息数量
        public int validCount; // 有效信息数量
    }

    public void init(){
        //暂时不需要做啥额外的初始化工作
        //以备后续扩展
    }

    //预定消息文件所在的目录和文件名
    //获取到指定 队列 对应的 消息文件 所在的 路径
    private String getQueueDir(String queueName){
        return "./data/"+queueName;
    }

    //获取该 队列的消息 数据文件路径
    // 二进制 使用txt 不太合适；
    // .bin  .dat :存储二进制
    private String getQueueDataPath(String queueName){
        return getQueueDir(queueName) + "/queue_data.txt";

    }

    //获取该队列的消息统计文件路径
    private  String getQueueStatPath(String queueName){
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    /**
    * 统计文件  读
    * @param queueName
    * @return
    */
    private Stat readStat (String queueName) throws IOException {
        //文本文件；使用Scanner
        Stat stat = new Stat();
        InputStream inputStream = null;
        try {
              inputStream = new FileInputStream(getQueueStatPath(queueName));
            Scanner scanner = new Scanner(inputStream);
           stat.totalCount =  scanner.nextInt();
           stat.validCount =  scanner.nextInt();
            return stat;
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }finally {
            inputStream.close();

        }
        return null;
    }
    /**
    * 统计文件  写
    * @param queueName
    * @return
    */
    private void writeStat(String queueName,Stat stat) throws IOException {
        // PrintWriter 写文件
        //outputStream 打开文件，默认情况下，会直接把源文件清空 ， 新的数据覆盖了旧的
        OutputStream outputStream = null;
        try {
              outputStream =  new FileOutputStream(getQueueStatPath(queueName));
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t"+stat.validCount);
            printWriter.flush();

        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }finally {
            outputStream.close();

        }

    }
    /**
    * 创建队列相应的文件和目录
    * @param queueName
    * @return
    */
    public void createQueueFiles(String queueName) throws Exception {
        //1. 创建队列相应的目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            //目录不存在：创建该目录
            boolean mkdirs = baseDir.mkdirs();
            if(!mkdirs){
                throw new Exception("目录创建失败，baseDir = "+baseDir.getAbsolutePath());
            }
        }

        //2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            boolean newFile = queueDataFile.createNewFile();
            if(!newFile){
                throw new IOException("创建文件失败 queueDataFile = "+queueDataFile.getAbsolutePath());
            }
        }
        //3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            boolean newFile = queueStatFile.createNewFile();
            if(!newFile){
                throw new IOException("创建文件失败 queueStatFile = "+queueStatFile.getAbsolutePath());
            }
        }

        //4. 给消息统计文件，设置初始值 0\t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName,stat);
    }

    /**
    * 删除队列的目录和文件
     * 队列删除的时候， 相应的目录和文件 也要删除
     * 先删除里面的文件，再删除目录
    * @param queueName
    * @return
    */
    public void destroyQueueFiles(String queueName) throws IOException {
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean deleteData = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean deleteStat =queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean deleteDir = baseDir.delete();
        if(!deleteData || !deleteStat || !deleteDir){
            // 有任意一个删除失败，都算整体删除失败
            throw new IOException("删除队列目录和文件失败！ baseDir = "+baseDir.getAbsolutePath());
        }
    }

    /**
    * 检查队列的目录和文件是否存在
     * 后续，有生产者 给broker server 生产消息了， 这个消息就可能就需要记录到文件上（取决于是否要持久化）
    * @param queueName
    * @return
    */
    public boolean checkFilesExists(String queueName){
        //判定队列的数据文件和统计文件是否都存在？
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            return false;
        }
        return true;
    }

   /**
   * 把一个新的消息，放到队列对应的文件中
   * @param queue 表示要把消息写入到队列
   * @param message 要写的消息
   * @return
   */
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1. 检查一下当前要写入的队列对应的文件是否存在
        if(!checkFilesExists(queue.getName())){
            throw new MqException("[MessageFileManager] 队列对应的文件不存在 ，queueName = "+queue.getName());
        }
        //2. 把Message 对象，进行序列化，转成二进制的字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        synchronized (queue){
            //3. 先获取到当前的队列数据文件的长度，计算出该 Message 的offsetBeg和 offsetEnd
            // 把新的Message 数据，写入到队列数据文件的末尾，此时Message对象的 offsetBeg ： 当前文件长度 +4
            // offsetEnd: 当前文件长度 +4+Message自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            // queueDataFile.length() 文件长度，单位字节
            message.setOffsetBeg(queueDataFile.length()+4);
            message.setOffsetEnd(queueDataFile.length()+4+messageBinary.length);
            //4. 写入消息到数据文件 ，追加写入
            try(OutputStream outputStream = new FileOutputStream(queueDataFile,true)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    //接下来要先写当前消息的长度，占据 4个字节的
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            //5.更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount+=1;
            stat.validCount+=1;
            writeStat(queue.getName(),stat);
        }

    }
    /**
    * 删除消息的方法
     * 逻辑删除
     * 就是把 isValid 设置为 0
    * @param queue
    * @param message 必须包含有效的 offsetBeg 和 offsetEnd
    * @return
    */
    public void deleteMessage(MSGQueue queue,Message message) throws IOException, ClassNotFoundException {
       synchronized (queue){
           //1. 先把 文件中这一段数据，读出来，
           try(RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()),"rw")){
               byte[] bufferSrc = new byte[(int)(message.getOffsetEnd() - message.getOffsetBeg())];
               randomAccessFile.seek(message.getOffsetBeg());
               randomAccessFile.read(bufferSrc);

               //2. 还原回 Message 对象 ,反序列化
               Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);

               //3. 把isValid 改成 0
               diskMessage.setIsValid((byte) 0x0);

               //4. 把上述数据重新写回到文件
               byte[] bufferDest = BinaryTool.toBytes(diskMessage);
               // 注意：虽然上面已经 seek过了，但是读写操作，都会影响 seek
               randomAccessFile.seek(message.getOffsetBeg());
               randomAccessFile.write(bufferDest);

           }
           //更新：统计文件
           Stat stat = readStat(queue.getName());
           if(stat.validCount >0){
               stat.validCount-=1;
           }
           writeStat(queue.getName(),stat);
       }


    }
    /**
    * 使用这个方法，从文件中，读取出所有的消息内容，加载到内存中（具体来说是放到一个链表里）
     * 这个方法，准备在 程序启动 的时候，进行调用
     *
    * @param queueName 这个方法不需要加锁，只需要使用queueName就够了
    * @return LinkedList<Message> ： 后序可以头删操作
    */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)) {
               long currentOffset = 0;
                //循环读取
                while(true){
                    //1. 读取消息的长度
                    // readInt 可能会读到文件末尾 ，会抛出一个  EOFException 异常
                    int messageSize =  dataInputStream.readInt();
                    //2. 按照这个长度 ，读取 消息长度
                    byte[] buffer = new byte[messageSize];
                    int actualSize =   dataInputStream.read(buffer);
                    if(messageSize != actualSize){
                        //如果不匹配，说明文件有问题，格式错乱了
                        throw new MqException("[MessageFIleManager] 文件格式错误 ,queueName = "+queueName);
                    }
                    //3. 把这个读到的二进制数据，反序列回 Message
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    //4. 检验 消息是否无效
                    if(message.getIsValid() != 0x1){
                        //无效数据。直接跳过
                        //虽然消息是无效的，但是offset需要更新
                        currentOffset += (4+messageSize);
                        continue;
                    }
                    //5. 如果是有效数据，加入链表中，
                    //    加入之前，需要填写 offsetBeg、offsetEnd
                    //    计算，OffsetBeg ， 需要光标位置，手动计算 光标
                    message.setOffsetBeg(currentOffset+4);
                    message.setOffsetEnd(currentOffset+4+messageSize);
                    currentOffset += (4+messageSize);
                    messages.add(message);
                }
            }catch (EOFException e){
                //这个异常 代表 ： 文件读到末尾了
                //也不需要做啥操作
                System.out.println("[MessageFileManager] 恢复Message 数据完成");

            }

        }
        return messages;

    }

    /**
    * 检查当前是否 要针对该队列的消息数据文件进行GC
    * @param queueName
    * @return
    */
    public boolean checkGC(String queueName) throws IOException {

        Stat stat = readStat(queueName);
        if(stat.totalCount > 2000 && stat.validCount*1.0/ stat.totalCount <0.5){
            return true;
        }
        return false;
    }

    /**
    * 约定新的文件
    *
    * @return
    */
    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    /**
    * 通过这个方法 ，真正执行消息数据文件的垃圾回收操作
     * 使用复制算法来完成
     * 创建一个新的文件，名字就是 queue_data_new.txt
     * 把之前的消息数据文件中的有效信息都读出来，写到新的文件中
     * 删除旧的文件，再把新的文件改名回 queue_data.txt
     * 同时，记得更新消息统计文件
    * @param queue
    * @return
    */
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        //进行 gc 的时候，是针对消息文件进行大洗牌，在这个过程中，其他线程不能针对该队列的消息队列有任何修改
        synchronized (queue){
            //统计一下 时间
            long gcBeg = System.currentTimeMillis();
            //1. 创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                //正常情况下，这个文件不应该存在，如果存在：上次，gc未完成
                throw new MqException("[MessageFileManager] gc时发现队列中 queue_data_new.txt 已存在，queueName = "+queue.getName());
            }
            boolean newFile = queueDataNewFile.createNewFile();
            if(!newFile){
                //失败
                throw new MqException("[MessageFileManager] 创建文件失败，queueDataNewFile = "+queueDataNewFile.getAbsolutePath());
            }
            //2. 从旧的文件中，读取出所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            //3. 把有效消息，写入到新文件中
            try(OutputStream outputStream =  new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for(Message message :messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //4. 删除旧的数据文件，并且把新的文件进行重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            boolean delete = queueDataOldFile.delete();
            if(!delete){
                throw new MqException("[MessageFileManager] 文件删除失败，queueDataOldFile = "+queueDataOldFile.getAbsolutePath());
            }
            //queue_data_new.txt -> queue_data.txt
            boolean b = queueDataNewFile.renameTo(queueDataOldFile);
            if(!b){
                throw new MqException("[MessageFileManager] 文件重命名失败，queueDataNewFile = "+queueDataNewFile.getAbsolutePath()+", queueDataOldFile = "+queueDataOldFile.getAbsolutePath());

            }
            //5. 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.validCount= messages.size();
            stat.totalCount= messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 完毕，queueName "+queue.getName()+", time = " + (gcEnd-gcEnd)+" ms");
        }
    }


}
